package com.flying.cattle.handler;

import com.flying.cattle.config.SpringUtils;
import com.flying.cattle.entity.WebSocketSender;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.web.reactive.socket.HandshakeInfo;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

@Component
public class EchoHandler implements WebSocketHandler {

	//redis监听
	private RedisMessageListenerContainer redisMessageListenerContainer =  SpringUtils.getBean(RedisMessageListenerContainer.class);
	@Override
	public Mono<Void> handle(final WebSocketSession session) {

		HandshakeInfo handshakeInfo = session.getHandshakeInfo();
		//获取url参数
		Map<String, String> queryMap = getQueryMap(handshakeInfo.getUri().getQuery());

		String id = queryMap.getOrDefault("topic", "defaultId");
		//获取客户端消息
		Mono<Void> input = session.receive().map(WebSocketMessage::getPayloadAsText).map(msg -> id + ": " + msg)
				.doOnNext(System.out::println).then();

		WebSocketSender webSocketSender = WebSocketSender.builder().build();

		ChannelTopic channelTopic = new ChannelTopic(id);
		//服务端发送消息，订阅redis
		Mono<Void> output = session.send(Flux.create(sink -> {
			webSocketSender.setSession(session);
			webSocketSender.setSink(sink);
			//订阅redis监听
			redisMessageListenerContainer.addMessageListener(webSocketSender, channelTopic);
		}));

		return Mono.zip(input, output).doFinally(signalType -> {
			// 端开连接时候 移除订阅  防止内存一直占用
			redisMessageListenerContainer.removeMessageListener(webSocketSender,channelTopic);
		}).then();
	}

	//用于获取url参数
	private Map<String, String> getQueryMap(String queryStr) {
		Map<String, String> queryMap = new HashMap<>();
		if (!StringUtils.isEmpty(queryStr)) {
			String[] queryParam = queryStr.split("&");
			Arrays.stream(queryParam).forEach(s -> {
				String[] kv = s.split("=", 2);
				String value = kv.length == 2 ? kv[1] : "";
				queryMap.put(kv[0], value);
			});
		}
		return queryMap;
	}
}
